package com.corpgovernment.common.mq.consumer;

import com.corpgovernment.common.apollo.CommonApollo;
import com.corpgovernment.common.base.BaseService;
import com.corpgovernment.common.constant.CommonConst;
import com.corpgovernment.common.holder.TenantContextHolder;
import com.corpgovernment.common.mq.MessageWrapper;
import com.corpgovernment.common.mq.enums.MessageBizTypeEnum;
import com.corpgovernment.common.utils.LogSplicingUtils;
import com.corpgovernment.common.utils.TenantUtils;
import com.ctrip.corp.obt.generic.core.context.TenantContext;
import com.ctrip.corp.obt.generic.core.context.TraceContext;
import com.ctrip.corp.obt.generic.core.event.EventSubscriber;
import com.ctrip.corp.obt.generic.core.event.TenantEvent;
import lombok.extern.slf4j.Slf4j;
import com.ctrip.corp.obt.generic.utils.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import com.ctrip.corp.obt.generic.utils.CollectionUtils;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author jhxue
 * @create 2021-12-17-16:24
 */
@Slf4j
public abstract class AbstractMultiTenantRocketMQConsumer extends BaseService implements EventSubscriber<TenantEvent> {

    @Autowired
    private CommonApollo commonApollo;

	/**
	 * 存储消费者对象
	 */
	public Map<String, DefaultMQPushConsumer> consumerMap = new ConcurrentHashMap<>();

    @Override
    public void onEvent(TenantEvent event) {
        initConsumerMap(event);
    }


    private void initConsumerMap(TenantEvent event) {
        if (!event.isAddEventType()) {
            return;
        }
        List<String> tenantList = Arrays.asList(event.getTenantIds());
        if (CollectionUtils.isEmpty(tenantList)) {
            CompletableFuture.runAsync(() -> {
            TenantContext.setTenantId(CommonConst.DEFAULT_TENANT_ID);
            consumerMap.computeIfAbsent(CommonConst.DEFAULT_TENANT_ID, s -> createConsumer());
            TenantContext.unset();
            });
        } else {
            tenantList.stream().parallel().forEach(tenantId -> {
                TenantContext.setTenantId(tenantId);
                consumerMap.computeIfAbsent(tenantId, s -> createConsumer());
                TenantContext.unset();
            });
        }
    }

    /**
     * 通过消费者信息创建消费者
     */
    public DefaultMQPushConsumer createConsumer() {
        StringBuilder logContext = new StringBuilder();
        try {
            LogSplicingUtils.addLogContext(logContext, "开始启动MQ消费者了");

            String namesrvAddr = commonApollo.getConsumerNamesrvAddr();
            LogSplicingUtils.addLogContext(logContext, "namesrvAddr：%s", namesrvAddr);
            if (Objects.equals("f", namesrvAddr)) {
                return null;
            }

            MessageBizTypeEnum bizTypeEnum = getMessageBizType();
            LogSplicingUtils.addLogContext(logContext, "bizTypeEnum：%s", bizTypeEnum);
            if (bizTypeEnum == null) {
                return null;
            }

            String accessKey = commonApollo.getConsumerAccessKey();
            String accessKeySecret = commonApollo.getConsumerAccessKeySecret();
            LogSplicingUtils.addLogContext(logContext, "accessKey：%s, accessKeySecret：%s", accessKey, accessKeySecret);

            AclClientRPCHook rpcHook = null;
            if (StringUtils.isNotBlank(accessKey) && StringUtils.isNotBlank(accessKeySecret)) {
                rpcHook = new AclClientRPCHook(new SessionCredentials(accessKey, accessKeySecret));
            }

            DefaultMQPushConsumer consumer;
            String tenantGroupName = TenantUtils.getTenantTopic(bizTypeEnum.getGroupName());
            if (rpcHook != null) {
                consumer = new DefaultMQPushConsumer(commonApollo.getProducerNamespace(), tenantGroupName, rpcHook, new AllocateMessageQueueAveragely(), true, null);
                consumer.setAccessChannel(AccessChannel.CLOUD);
                LogSplicingUtils.addLogContext(logContext, "namespace：%s", commonApollo.getProducerNamespace());
                LogSplicingUtils.addLogContext(logContext, "rpcHook not null");
            } else {
                consumer = new DefaultMQPushConsumer(tenantGroupName);
                LogSplicingUtils.addLogContext(logContext, "rpcHook null");
            }
            // 此处设置instanceName，以防某租户topic下线影响其他客户
            consumer.setInstanceName(TenantContext.getTenantId());
            consumer.setNamesrvAddr(namesrvAddr);
            consumer.setConsumeThreadMin(commonApollo.getConsumeThreadMin());
            consumer.setConsumeThreadMax(commonApollo.getConsumeThreadMax());
            consumer.registerMessageListener(messageListener());
            //设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
            //如果非第一次启动，那么按照上次消费的位置继续消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //设置一次消费消息的条数，默认为1条
            consumer.setConsumeMessageBatchMaxSize(commonApollo.getConsumeMessageBatchMaxSize());
            String tenantTopic = TenantUtils.getTenantTopic(bizTypeEnum.getTopic());
            LogSplicingUtils.addLogContext(logContext, "tenantTopic: %s", tenantTopic);

            consumer.subscribe(tenantTopic, bizTypeEnum.getTags());
            consumer.start();
            LogSplicingUtils.addLogContext(logContext, "启动完成");
            return consumer;
        } catch (Exception e) {
            log.error("启动消费者者发生异常：{}", e);
        } finally {
            log.info("{}", logContext.toString());
        }
        return null;
    }

    protected MessageListenerConcurrently messageListener() {
        MessageBizTypeEnum bizTypeEnum = getMessageBizType();
        String tenantTopic = TenantUtils.getTenantTopic(bizTypeEnum.getTopic());
        String tenantId = TenantContextHolder.getTenantId();
        return (var1, var2) -> {
            if (!CollectionUtils.isEmpty(var1)) {
                for (MessageExt messageExt : var1) {
                    if (messageExt.getTopic().equals(tenantTopic)) {
                        try {
                            TenantContext.setTenantId(tenantId);
                            MessageWrapper messageWrapper = new MessageWrapper(messageExt);
                            String clientId = messageWrapper.getClientId();
                            Optional.ofNullable(clientId).ifPresent(TenantContext::setClientId);
                            TraceContext.setRequestId(String.valueOf(UUID.randomUUID()));
                            this.consume(messageExt, var2);
                        } catch (Exception e) {
                            log.warn("消费topic:{},失败，消息体：{}，打印异常：{}", tenantTopic, messageExt, e);
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                        } finally {
                            TenantContext.unset();
                        }
                    } else {
                        // 如果没有return success, consumer会重新消费该消息, 直到return success
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        };
    }

    /**
     * 消息消费具体逻辑
     *
     * @param messageExt
     * @param context
     */
    protected abstract void consume(MessageExt messageExt, ConsumeConcurrentlyContext context);

    /**
     * 获取当前consumer的messageBizType
     *
     * @return
     */
    protected abstract MessageBizTypeEnum getMessageBizType();
}
